package cn.doitedu.demo1;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/1/21
 * @Desc: 学大数据，上多易教育
 *   针对  活跃等级为 C 的用户
 *   在规则上线后，如果这些用户 发生了 X行为 >=3 次
 *   等他再发生  Y行为的时候，触发营销消息
 *
 *
 *   针对  价值等级为 A 的用户
 *   在规则上线后，如果这些用户 发生了 W行为 >=2 次
 *   等他再发生 K行为的时候，触发营销消息
 *
 **/
public class Demo1 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);


        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("ggg111")
                .setClientIdPrefix("ccc111")
                .setTopics("dwd-events")
                .build();


        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        SingleOutputStreamOperator<UserEvent> eventStream = streamSource.map(json -> JSON.parseObject(json, UserEvent.class));


        SingleOutputStreamOperator<Message> messages = eventStream.keyBy(UserEvent::getUser_id)
                .process(new KeyedProcessFunction<Long, UserEvent, Message>() {


                    HashMap<String, RuleCalculator> ruleCalculatorPool = new HashMap<>();

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        RuntimeContext runtimeContext = getRuntimeContext();

                        // 构造规则的运算机实例对象
                        String rule1_param = "{\n" +
                                "  \"rule_id\":\"rule-1\",\n" +
                                "  \"static_profile_condition\":{\n" +
                                "    \"tag_name\":\"tag0103\",\n" +
                                "    \"tag_value\":\"C\"\n" +
                                "  },\n" +
                                "  \"realtime_profile_condition\":{\n" +
                                "    \"event_id\":\"X\",\n" +
                                "    \"min_count\": 3\n" +
                                "  },\n" +
                                "  \"trigger_condition\":{\n" +
                                "    \"event_id\":\"Y\"\n" +
                                "  }\n" +
                                "}";
                        RuleCalculator rule1Calculator = new Rule_1_Calculator();
                        rule1Calculator.init(runtimeContext,rule1_param);
                        ruleCalculatorPool.put("rule-1",rule1Calculator);


                        String rule2_param = "{\n" +
                                "  \"rule_id\":\"rule-2\",\n" +
                                "  \"static_profile_condition\":{\n" +
                                "    \"tag_name\":\"tag0204\",\n" +
                                "    \"tag_value\":\"A\"\n" +
                                "  },\n" +
                                "  \"realtime_profile_condition\":{\n" +
                                "    \"event_id\":\"W\",\n" +
                                "    \"min_count\": 2\n" +
                                "  },\n" +
                                "  \"trigger_condition\":{\n" +
                                "    \"event_id\":\"K\"\n" +
                                "  }\n" +
                                "}";
                        RuleCalculator rule2Calculator = new Rule_1_Calculator();
                        rule2Calculator.init(runtimeContext,rule2_param);
                        ruleCalculatorPool.put("rule-2",rule2Calculator);



                    }

                    @Override
                    public void processElement(UserEvent userEvent, KeyedProcessFunction<Long, UserEvent, Message>.Context ctx, Collector<Message> out) throws Exception {

                        for (Map.Entry<String, RuleCalculator> entry : ruleCalculatorPool.entrySet()) {
                            String ruleId = entry.getKey();
                            RuleCalculator ruleCalculator = entry.getValue();

                            ruleCalculator.calc(userEvent,out);
                        }





                    }
                });

        messages.print();

        env.execute();

    }
}
